package cn.hesion;
/**
 * ClassName: WindowDemo <br/>
 * Description: <br/>
 * date: 2021/9/8 18:58<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import jdk.internal.util.xml.impl.Input;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.Random;

/**
 * 窗口机制
 * @author: hesion
 * @create: 2021-09-08 18:58
 **/
public class WindowDemo {
    public static void main(String[] args) throws Exception {
        //获取数据流 data
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //自定义数据源 一秒发送一个递增的数据
        DataStreamSource<String> data = env.addSource(new SourceFunction<String>() {
            int num = 0;

            @Override
            public void run(SourceContext<String> sourceContext) throws Exception {
                while (true) {
                    sourceContext.collect(String.valueOf(num));
                    num++;
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        //2 获取窗口
        SingleOutputStreamOperator<Tuple3<String, String, String>> maped = data.map(new MapFunction<String, Tuple3<String, String, String>>() {

            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                long l = System.currentTimeMillis();
                String dateTime = sdf.format(l);
                Random random = new Random();
                int randomNum = random.nextInt(5);
                return new Tuple3<>(value, dateTime, String.valueOf(randomNum));
            }
        });

        KeyedStream<Tuple3<String, String, String>, String> keybyed = maped.keyBy(value -> value.f0);

        WindowedStream<Tuple3<String, String, String>, String, TimeWindow> timeWindow = keybyed.timeWindow(Time.seconds(5));

        SingleOutputStreamOperator<String> applyed = timeWindow.apply(new WindowFunction<Tuple3<String, String, String>, String, String, TimeWindow>() {

            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<Tuple3<String, String, String>> input, Collector<String> collector) throws Exception {
                Iterator<Tuple3<String, String, String>> iterator = input.iterator();
                StringBuffer sb = new StringBuffer();
                System.out.println("........");
                while (iterator.hasNext()) {
                    Tuple3<String, String, String> next = iterator.next();
                    sb.append(next.f0 + "....." + next.f1 + "....." + next.f2);
                }
                String s1 = s + "..." + sdf.format(timeWindow.getStart()) + "...." + sdf.format(timeWindow.getEnd()) + sb;

                collector.collect(s1);

            }
        });

        applyed.print();
        env.execute();
    }
}
